package com.alison.datastream.etl;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.Collector;

import java.util.Collections;
import java.util.Map;

/**
 * @Author alison
 * @Date 2024/4/4 0:20
 * @Version 1.0
 * @Description
 */
public class StateApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> rides = env.fromElements("1", "1", "2", "3", "4");
        rides.keyBy(String::valueOf).flatMap(new DeduplicateFlatMapFunction()).print();
        env.execute();
    }

    //自定义实现richFunction 因为rich才可以访问上下文
    public static class DeduplicateFlatMapFunction extends RichFlatMapFunction<String, String> {
        ValueState<Map<String, Boolean>> keyHasBeanSeenState;

        @Override
        public void open(Configuration parameters) throws Exception {
            ValueStateDescriptor keyHasDescri = new ValueStateDescriptor("hasBeanSeenState", Map.class);
            keyHasBeanSeenState = super.getRuntimeContext().getState(keyHasDescri);
        }

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            Map<String, Boolean> map = keyHasBeanSeenState.value();
            if (CollectionUtil.isNullOrEmpty(map) || !map.containsKey(value)) {
                out.collect(value);
                keyHasBeanSeenState.update(Collections.singletonMap(value, true));
            } else {
                System.out.println("之前已经遇到过" + value + "不存储");
            }
        }
    }
}
